package controller

import (
	"context"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"mig2kwdb/common"
	"mig2kwdb/readers"
	"mig2kwdb/writers"
	"reflect"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/araddon/dateparse"
	"github.com/fatih/color"
)

// Controller 控制器结构体
type Controller struct {
	reader         readers.Reader
	writer         writers.Writer
	task           *common.MigrateTask
	batchSize      int
	parallelTable  int
	parallelInsert int
}

// NewController 创建新的控制器实例
func NewController(reader readers.Reader, writer writers.Writer) *Controller {
	return &Controller{
		reader: reader,
		writer: writer,
	}
}

// formatValue 根据数据类型格式化值
func formatValue(value interface{}, dataType string) string {
	if sqlValue, ok := value.(*sql.NullString); ok && sqlValue.Valid {
		// 根据数据类型决定是否添加引号
		switch strings.ToUpper(dataType) {
		case "TINYINT", "SMALLINT", "MEDIUMINT", "INT", "INTEGER", "BIGINT",
			"DECIMAL", "NUMERIC", "FLOAT", "DOUBLE", "REAL":
			return sqlValue.String
		default:
			// 默认作为字符串处理
			return fmt.Sprintf("'%s'", sqlValue.String)
		}
	}
	return "NULL"
}
func buildMultiColumnCondition(baseCondition string, lastValue []interface{}, columns []common.KwdbColumn, operator string) string {
	type KeyInfo struct {
		Column common.KwdbColumn
		Value  interface{}
	}

	var primaryKeys []KeyInfo
	for i, col := range columns {
		if col.PrimaryKey.String == "YES" {
			primaryKeys = append(primaryKeys, KeyInfo{
				Column: col,
				Value:  lastValue[i],
			})
		}
	}
	// 如果没有主键，则使用UNIQUE
	if len(primaryKeys) == 0 {
		for i, col := range columns {
			if col.Unique.String == "YES" {
				primaryKeys = append(primaryKeys, KeyInfo{
					Column: col,
					Value:  lastValue[i],
				})
			}
		}
	}
	if len(primaryKeys) == 0 {
		return baseCondition
	}

	var conditions []string

	for i := 0; i < len(primaryKeys); i++ {
		var conditionParts []string

		for j := 0; j < i; j++ {
			conditionParts = append(conditionParts,
				fmt.Sprintf("%s = %s",
					primaryKeys[j].Column.Name,
					formatValue(primaryKeys[j].Value, primaryKeys[j].Column.Type)))
		}

		conditionParts = append(conditionParts,
			fmt.Sprintf("%s %s %s",
				primaryKeys[i].Column.Name,
				operator,
				formatValue(primaryKeys[i].Value, primaryKeys[i].Column.Type)))

		branchCondition := "(" + strings.Join(conditionParts, " AND ") + ")"
		conditions = append(conditions, branchCondition)
	}

	finalCondition := strings.Join(conditions, " OR ")

	if baseCondition != "" {
		return fmt.Sprintf("(%s) AND (%s)", baseCondition, finalCondition)
	}
	return finalCondition
}

func (c *Controller) GetTableColumn(databaseName, schemaName, tableName string) ([]common.KwdbColumn, error) {
	if c.reader == nil {
		return nil, errors.New("reader is nil")
	}
	return c.reader.GetTableColumn(databaseName, schemaName, tableName)
}

// retryInsertWithBackoff 带重试机制的插入操作
func (c *Controller) retryInsertWithBackoff(targetDB, targetSchema, targetTableName string,
	targetColumns []common.KwdbColumn, currentValues [][]interface{}, IsSeries bool) error {

	maxRetries := 50
	for attempt := 0; attempt < maxRetries; attempt++ {
		_, insertErr := c.writer.InsertUpdateRecords(targetDB, targetSchema, targetTableName,
			targetColumns, currentValues, IsSeries)
		if insertErr == nil {
			return nil // 成功则返回
		}

		if c.writer.IsNetworkError(insertErr) {
			if attempt < maxRetries-1 { // 不是最后一次尝试
				time.Sleep(time.Duration(attempt+1) * time.Second) // 指数退避
				continue                                           // 重试
			}
		}
		// 非网络错误或重试次数用完，记录错误并返回
		log.Printf("Insert failed after %d attempts: %v", maxRetries, insertErr)
		return insertErr
	}
	return nil
}
func (c *Controller) SyncOneTable(ctx context.Context, stats *common.TaskStattistics, sourceDB, sourceSchema string, sourceTable common.TableConfig, targetDB, targetSchema string, targetTable common.TableConfig, IsSeries bool) error {
	log.Println("Syncing table:", sourceDB, sourceSchema, sourceTable.Name)
	tableStats := common.TableStatistics{}

	tableDef, err := c.reader.GetTableDefinition(sourceDB, sourceSchema, sourceTable.Name)
	if err != nil {
		return err
	}
	columns := tableDef.Columns

	hasKey := false
	for _, col := range columns {
		if col.PrimaryKey.String == "YES" {
			hasKey = true
			break
		} else if col.Unique.String == "YES" {
			hasKey = true
			break
		}
	}
	tableStats.StartTime = time.Now()
	defer func() {
		tableStats.DBName = sourceDB
		tableStats.SchemaName = sourceSchema
		tableStats.TableName = sourceTable.Name
		tableStats.EndTime = time.Now()

		if err != nil {
			tableStats.Error = err.Error()
		}
		stats.TableStats = append(stats.TableStats, tableStats)
	}()
	targetColumns := make([]common.KwdbColumn, len(columns))
	copy(targetColumns, columns)
	// 只有当targetTable.Includes为空时才创建表
	// 如果不为空，表示指定了需要的列，则表示表已经存在
	if len(targetTable.Includes) == 0 {
		if !targetTable.UseExist {
			c.writer.DropTable(targetDB, targetSchema, targetTable.Name, IsSeries)
			if err = c.writer.CreateTable(targetDB, targetSchema, targetTable.Name, tableDef, IsSeries); err != nil {
				return err
			}
		}
	} else {
		if len(targetTable.Includes) != len(columns) {
			log.Println("targetTable.Includes and sourceTable.Includes are not equal", sourceTable.Name, targetTable.Name)
			return errors.New("targetTable.Includes and sourceTable.Includes are not equal")
		}
		// 更新列名
		for i, name := range targetTable.Includes {
			targetColumns[i].Name = name
		}
	}
	retryMaxTimes := 100
	lastValue := make([]interface{}, len(columns))
	needResume := false
	lastValueAssigned := false

	// goroutine池相关变量
	maxConcurrentInsertions := c.parallelInsert
	insertionQueue := make(chan func() error, maxConcurrentInsertions*2) // 缓冲队列
	insertionErrors := make(chan error, maxConcurrentInsertions*2)
	var wg sync.WaitGroup
	ctxWithCancel, cancel := context.WithCancel(ctx)
	defer cancel()

	// 启动worker池
	for i := 0; i < maxConcurrentInsertions; i++ {
		wg.Add(1)
		go func(workerID int) {
			defer wg.Done()
			for {
				select {
				case <-ctxWithCancel.Done():
					return
				case insertFunc, ok := <-insertionQueue:
					if !ok {
						return
					}
					if err := insertFunc(); err != nil {
						select {
						case insertionErrors <- err:
						default:

						}
						cancel() // 取消所有goroutine
					}
				}
			}
		}(i)
	}

	// 等待所有插入完成并处理错误的函数
	waitForInsertions := func() error {
		// 等待所有已提交的任务完成
		close(insertionQueue)
		wg.Wait()

		select {
		case err := <-insertionErrors:
			return err
		default:
			return nil
		}
	}

	var totalNum int64 = 0
	for i := 0; i < retryMaxTimes; i++ {
		if common.IsStopped(ctx) {
			cancel()
			waitForInsertions()
			return nil
		}
		condition := sourceTable.Condition
		if needResume && lastValueAssigned {
			if !hasKey {
				//truncate the target table
				_, err = c.writer.GetDB().Exec(fmt.Sprintf("TRUNCATE TABLE %s.%s.%s", targetDB, targetSchema, targetTable.Name))
				if err != nil {
					if c.writer.IsNetworkError(err) {
						fmt.Println()
						log.Println("Disconnected, reconnecting...")
						time.Sleep(time.Duration(i+1) * time.Second)
						needResume = true
						continue
					}
				}
			} else {
				condition = buildMultiColumnCondition(condition, lastValue, columns, ">")
			}
		}
		if condition != "" {
			log.Println("condition:", condition)
		}
		rows, errn := c.reader.GetConditionRows(sourceDB, sourceSchema, sourceTable.Name, columns, condition)
		if errn != nil {
			if c.reader.IsNetworkError(errn) {
				fmt.Println()
				log.Println("Disconnected, reconnecting...")
				time.Sleep(time.Duration(i+1) * time.Second)
				needResume = true
				continue
			}
			log.Println(errn)
			err = errn
			return errn
		}
		defer rows.Close()
		var values [][]interface{}
		num := 0
		var oneChunkSize int64 = 1024 * 1024 * 10
		var totalChunkSize int64 = 0
		if !hasKey {
			tableStats.Count = 0
		}
		for rows.Next() {
			if common.IsStopped(ctx) {
				cancel()
				waitForInsertions()
				return nil
			}
			value := make([]interface{}, len(columns))
			for i := 0; i < len(columns); i++ {
				value[i] = &sql.NullString{}
			}
			err = rows.Scan(value...)
			if err != nil {
				if c.reader.IsNetworkError(err) {
					fmt.Println()
					log.Println("Disconnected, reconnecting...")
					time.Sleep(time.Duration(i+1) * time.Second)
					needResume = true
					break
				}
				log.Println(err)
				cancel()
				waitForInsertions()
				return err
			}
			for _, v := range value {
				sqlValue := v.(*sql.NullString)
				if sqlValue.Valid {
					totalChunkSize += int64(len(sqlValue.String))
				}
			}

			num++
			totalNum++
			values = append(values, value)
			if num >= c.batchSize || totalChunkSize >= oneChunkSize {
				// 使用goroutine异步插入
				currentValues := make([][]interface{}, len(values))
				for i, row := range values {
					currentValues[i] = make([]interface{}, len(row))
					for j, val := range row {
						if nullString, ok := val.(*sql.NullString); ok {
							// 创建新的 *sql.NullString 实例
							currentValues[i][j] = &sql.NullString{
								String: nullString.String,
								Valid:  nullString.Valid,
							}
						} else {
							currentValues[i][j] = val
						}
					}
				}

				select {
				case <-ctxWithCancel.Done():
					// 上下文已取消，退出
					waitForInsertions()
					return ctxWithCancel.Err()
				case insertionQueue <- func() error {
					return c.retryInsertWithBackoff(targetDB, targetSchema, targetTable.Name, targetColumns, currentValues, IsSeries)
				}:
					// 成功提交到队列
				case err := <-insertionErrors:
					// 有错误发生
					cancel()
					waitForInsertions()
					return err
				}

				num = 0
				totalChunkSize = 0
				copy(lastValue, value)
				lastValueAssigned = true
				values = values[0:0]
				fmt.Printf("\r[%s] Completed rows 已完成记录数: %d", common.GetFullTableName(sourceDB, sourceSchema, sourceTable.Name), totalNum)
			}
			i = 0
			needResume = false
		}
		if needResume {
			continue
		}
		err = rows.Err()
		if err != nil {
			if c.reader.IsNetworkError(err) {
				fmt.Println()
				log.Println("Disconnected, reconnecting...")
				time.Sleep(time.Duration(i+1) * time.Second)
				needResume = true
				continue
			}
			log.Println(err)
			cancel()
			waitForInsertions()
			return err
		}
		if len(values) > 0 {
			// 处理最后一批数据
			currentValues := make([][]interface{}, len(values))
			copy(currentValues, values)

			select {
			case <-ctxWithCancel.Done():
				waitForInsertions()
				return ctxWithCancel.Err()
			case insertionQueue <- func() error {
				return c.retryInsertWithBackoff(targetDB, targetSchema, targetTable.Name, targetColumns, currentValues, IsSeries)
			}:
			case err := <-insertionErrors:
				cancel()
				waitForInsertions()
				return err
			}

			fmt.Printf("\r[%s] Completed rows 已完成记录数: %d", common.GetFullTableName(sourceDB, sourceSchema, sourceTable.Name), totalNum)
		}

		// 等待所有插入操作完成
		if err := waitForInsertions(); err != nil {
			return err
		}

		tableStats.Count += totalNum
		fmt.Println()
		break
	}
	return err
}
func (c *Controller) GetTableColumns(sourceDB, sourceSchema string, sourceTable common.TableConfig, tabldeDef common.TableDefinition) ([]common.KwdbColumn, error) {
	//columns, err := c.reader.GetTableColumn(sourceDB, sourceSchema, sourceTable.Name)
	columns := tabldeDef.Columns

	if len(sourceTable.Excludes) > 0 {
		excludeMap := make(map[string]bool)
		for _, exclude := range sourceTable.Excludes {
			excludeMap[exclude] = true
		}

		// 过滤列
		filteredColumns := []common.KwdbColumn{}
		for _, column := range columns {
			if !excludeMap[column.Name] {
				filteredColumns = append(filteredColumns, column)
			}
		}
		columns = filteredColumns

	}
	if len(sourceTable.Includes) > 0 {
		includeMap := make(map[string]bool)
		for _, include := range sourceTable.Includes {
			includeMap[include] = true
		}

		//保证顺序和sourceTable.Includes一致
		filteredColumns := []common.KwdbColumn{}
		for _, includeName := range sourceTable.Includes {
			for _, column := range columns {
				if column.Name == includeName {
					filteredColumns = append(filteredColumns, column)
					break
				}
			}
		}
		columns = filteredColumns
	}
	return columns, nil
}
func (c *Controller) CompareValue(dataType string, sourceValue, targetValue interface{}) int {
	// 类型断言为 *sql.NullString
	sourceNs, sourceOk := sourceValue.(*sql.NullString)
	targetNs, targetOk := targetValue.(*sql.NullString)

	// 如果类型断言失败，则跳过该列
	if !sourceOk || !targetOk {
		return 0
	}
	if !sourceNs.Valid && !targetNs.Valid {
		return 0
	}
	if sourceNs.String == targetNs.String {
		return 0
	}
	if !sourceNs.Valid {
		if strings.ToUpper(dataType) == "BLOB" {
			//KWDB的blob类型，如果为空，则为\x
			if targetNs.String == "\\x" {
				return 0
			}
		}
		return -1
	}
	if !targetNs.Valid {
		return 1
	}
	switch strings.ToUpper(dataType) {
	case "BOOL":
		sourceBool, err1 := strconv.ParseBool(sourceNs.String)
		if err1 != nil {
			return -1
		}
		targetBool, err2 := strconv.ParseBool(targetNs.String)
		if err2 != nil {
			return 1
		}
		if sourceBool == targetBool {
			return 0
		}

	case "DATE":
		ts1, err := time.Parse("15:04:05", sourceNs.String)
		if err != nil {
			ts1, _ = time.Parse(time.RFC3339, sourceNs.String)
		}
		ts2, _ := time.Parse(time.RFC3339, targetNs.String)

		if ts1.Equal(ts2) {
			return 0
		}
		if ts1.Year() == ts2.Year() && ts1.Month() == ts2.Month() && ts1.Day() == ts2.Day() {
			return 0
		}
	case "TIME":
		ts1, err := time.Parse("15:04:05", sourceNs.String)
		if err != nil {
			ts1, _ = time.Parse(time.RFC3339, sourceNs.String)
		}
		ts2, _ := time.Parse(time.RFC3339, targetNs.String)
		//由于KWDB时间精确到毫秒，所以在毫秒级别进行比较
		ts1 = ts1.Truncate(time.Millisecond)
		ts2 = ts2.Truncate(time.Millisecond)
		if ts1.Equal(ts2) {
			return 0
		}
		if ts1.Hour() == ts2.Hour() && ts1.Minute() == ts2.Minute() &&
			ts1.Second() == ts2.Second() && ts1.Nanosecond()/1000 == ts2.Nanosecond()/1000 {
			return 0
		}
	case "TIMESTAMP", "TIMESTAMPTZ", "TIMESTAMP WITH TIME ZONE":
		if "TIMESTAMP" == strings.ToUpper(dataType) {
			ts1, err1 := dateparse.ParseAny(sourceNs.String)
			if err1 != nil {
				return -1
			}
			ts2, err2 := dateparse.ParseAny(targetNs.String)
			if err2 != nil {
				return 1
			}

			if ts1.Equal(ts2) {
				return 0
			}
		}
		ts1, _ := time.Parse(time.RFC3339, sourceNs.String)
		ts2, _ := time.Parse(time.RFC3339, targetNs.String)
		//由于KWDB时间精确到毫秒，所以在毫秒级别进行比较
		ts1 = ts1.Truncate(time.Millisecond)
		ts2 = ts2.Truncate(time.Millisecond)
		if ts1.Equal(ts2) {
			return 0
		}
	case "JSON", "JSONB":
		var obj1, obj2 interface{}
		if err := json.Unmarshal([]byte(sourceNs.String), &obj1); err != nil {
			return -1
		}
		if err := json.Unmarshal([]byte(targetNs.String), &obj2); err != nil {
			return 1
		}
		if reflect.DeepEqual(obj1, obj2) {
			return 0
		}
		return strings.Compare(sourceNs.String, targetNs.String)
	case "DOUBLE", "FLOAT", "FLOAT4", "FLOAT8":
		f1, _ := strconv.ParseFloat(sourceNs.String, 64)
		f2, _ := strconv.ParseFloat(targetNs.String, 64)
		if f1 == f2 {
			return 0
		}
	case "VARBINARY", "BINARY":
		var s1 string
		var s2 string
		if !strings.HasPrefix(sourceNs.String, "0x") {
			s1 = hex.EncodeToString([]byte(sourceNs.String))
		} else {
			s1 = sourceNs.String[2:]
		}
		if !strings.HasPrefix(targetNs.String, "0x") {
			s2 = hex.EncodeToString([]byte(targetNs.String))
		} else {
			s2 = targetNs.String[2:]
		}
		return strings.Compare(s1, s2)

	case "BYTEA", "BLOB":
		s1 := hex.EncodeToString([]byte(sourceNs.String))
		s2 := targetNs.String
		if strings.HasPrefix(s2, "\\x") {
			s2 = s2[2:]
		}
		return strings.Compare(s1, s2)

	}

	// 两者都有效，比较字符串值
	if sourceNs.String < targetNs.String {
		return -1
	} else if sourceNs.String > targetNs.String {
		return 1
	}
	return 0
}

// CompareData 比较源表和目标表当前行数据的主键和其他字段
// 返回值:
//
//	0: 所有字段都相同
//	1: 主键相同，但其他字段不同
//	2: 源表主键大于目标表主键
//	3: 源表主键小于目标表主键
func (c *Controller) CompareData(columns []common.KwdbColumn, sourceValue, targetValue []interface{}) int {
	// 查找主键列索引
	var primaryKeyIndices []int
	var nonKeyIndices []int

	for i, col := range columns {
		if col.PrimaryKey.String == "YES" || col.Unique.String == "YES" {
			primaryKeyIndices = append(primaryKeyIndices, i)
		} else {
			nonKeyIndices = append(nonKeyIndices, i)
		}
	}

	// 首先比较主键
	keyComparison := 0
	for _, pkIndex := range primaryKeyIndices {
		sourceKey := sourceValue[pkIndex]
		targetKey := targetValue[pkIndex]
		keyComparison = c.CompareValue(columns[pkIndex].Type, sourceKey, targetKey)
		if keyComparison != 0 {
			break
		}
		// 如果相等，继续比较下一个主键
	}

	// 根据主键比较结果返回
	if keyComparison > 0 {
		return 2 // 源表主键大于目标表主键
	} else if keyComparison < 0 {
		return 3 // 源表主键小于目标表主键
	}

	// 主键相同，检查其他字段是否相同
	for _, nonKeyIndex := range nonKeyIndices {
		sourceField := sourceValue[nonKeyIndex]
		targetField := targetValue[nonKeyIndex]

		if c.CompareValue(columns[nonKeyIndex].Type, sourceField, targetField) != 0 {
			return 1
		}
	}

	// 所有字段都相同
	return 0
}
func (c *Controller) VerifyOneTable(ctx context.Context, stats *common.TaskStattistics, sourceDB, sourceSchema string, sourceTable common.TableConfig, targetDB, targetSchema string, targetTable common.TableConfig, IsSeries bool) error {
	log.Println("Verifying table", sourceTable.Name)
	tableStats := common.TableStatistics{}
	tableDef, errres := c.reader.GetTableDefinition(sourceDB, sourceSchema, sourceTable.Name)
	if errres != nil {
		return errres
	}
	columns := tableDef.Columns
	hasKey := false
	for _, col := range columns {
		if col.PrimaryKey.String == "YES" {
			hasKey = true
			break
		} else if col.Unique.String == "YES" {
			hasKey = true
			break
		}
	}
	tableStats.StartTime = time.Now()
	defer func() {
		tableStats.DBName = sourceDB
		tableStats.SchemaName = sourceSchema
		tableStats.TableName = sourceTable.Name
		tableStats.EndTime = time.Now()
		if errres != nil {
			tableStats.Error = errres.Error()
		}
		stats.TableStats = append(stats.TableStats, tableStats)

	}()
	if !hasKey {
		//log.Println("Table", sourceTable.Name, "has no primary key, skip the verify process")
		tableName := sourceTable.Name
		if sourceSchema != "" {
			tableName = sourceSchema + "." + tableName
		}
		if sourceDB != "" {
			tableName = sourceDB + "." + tableName
		}
		return errors.New("Table " + tableName + " has no primary key, skip the verify process")
	}
	targetColumns := make([]common.KwdbColumn, len(columns))
	copy(targetColumns, columns)
	// 只有当targetTable.Includes为空时才创建表
	// 如果不为空，表示指定了需要的列，则表示表已经存在
	if len(targetTable.Includes) > 0 {
		if len(targetTable.Includes) != len(columns) {
			log.Println("targetTable.Includes and sourceTable.Includes are not equal", sourceTable.Name, targetTable.Name)
			err := errors.New("targetTable.Includes and sourceTable.Includes are not equal")
			errres = err
			return err
		}
		// 更新列名
		for i, name := range targetTable.Includes {
			targetColumns[i].Name = name
		}
	}
	//比较数据
	var added, updated, deleted int64 = 0, 0, 0
	condition := sourceTable.Condition
	if condition != "" {
		log.Println("condition:", condition)
	}
	rowsSource, err := c.reader.GetConditionRows(sourceDB, sourceSchema, sourceTable.Name, columns, condition)
	if err != nil {
		log.Println(err)
		errres = err
		return err
	}
	defer rowsSource.Close()

	rowsTarget, err := c.writer.GetConditionRows(targetDB, targetSchema, targetTable.Name, targetColumns, condition)
	if err != nil {
		log.Println(err)
		errres = err
		return err
	}
	defer rowsTarget.Close()

	sourceHasData := rowsSource.Next()
	targetHasData := rowsTarget.Next()
	var totalRows int64 = 0
	for sourceHasData && targetHasData {
		if common.IsStopped(ctx) {
			return nil
		}
		sourceValue := make([]interface{}, len(columns))
		targetValue := make([]interface{}, len(columns))
		for i := 0; i < len(columns); i++ {

			sourceValue[i] = &sql.NullString{}
			targetValue[i] = &sql.NullString{}

		}
		err := rowsSource.Scan(sourceValue...)
		if err != nil {
			errres = err
			log.Println(err)
			break
		}
		err = rowsTarget.Scan(targetValue...)
		if err != nil {
			log.Println(err)
			errres = err
			break
		}
		totalRows++
		// compare the keys
		diffRes := c.CompareData(targetColumns, sourceValue, targetValue)
		if diffRes == 0 {
			sourceHasData = rowsSource.Next()
			targetHasData = rowsTarget.Next()
			continue
		}
		if diffRes == 1 { //key match,but other is different
			sourceHasData = rowsSource.Next()
			targetHasData = rowsTarget.Next()
			//update target
			//log.Println("Update target")
			_, err := c.writer.UpdateRecord(targetDB, targetSchema, targetTable.Name, targetColumns, sourceValue, "sync", IsSeries)
			if err != nil {
				log.Println(err)
				errres = err
				break
			}
			updated++
		} else if diffRes == 2 { // source key is greater than dest
			targetHasData = rowsTarget.Next()
			//delete the target data
			//log.Println("Delete target")
			_, err := c.writer.DeleteRecord(targetDB, targetSchema, targetTable.Name, targetColumns, targetValue, "sync", IsSeries)
			if err != nil {
				log.Println(err)
				errres = err
				break
			}
			deleted++
		} else { // source key is less than dest
			sourceHasData = rowsSource.Next()
			//add the source data
			//log.Println("Add source")
			_, err := c.writer.InsertRecords(targetDB, targetSchema, targetTable.Name, targetColumns, [][]interface{}{sourceValue}, "sync", IsSeries)
			if err != nil {
				log.Println(err)
				errres = err
				break
			}
			added++
		}
	}
	if sourceHasData {
		log.Println("Source has more data")
		// insert the rest data
		count, err := c.InsertRemainingSourceData(ctx, rowsSource, targetDB, targetSchema, targetTable.Name, targetColumns, IsSeries)
		if err != nil {
			log.Println(err)
			errres = err

		}
		added += count
		totalRows += count
	}
	if targetHasData {
		log.Println("Target has more data")
		// delete the rest data
		count, err := c.DeleteRemainingTargetData(ctx, rowsTarget, targetDB, targetSchema, targetTable.Name, targetColumns, IsSeries)
		if err != nil {
			log.Println(err)
			errres = err
		}
		deleted += count
	}
	log.Println("Verify finished")
	if added > 0 || updated > 0 || deleted > 0 {
		log.Println("Added Records:", added, ",Updated Records:", updated, ",Deleted Records:", deleted)
	} else {
		log.Println("No data changed")
	}
	tableStats.AddedRows = added
	tableStats.UpdatedRows = updated
	tableStats.DeletedRows = deleted
	tableStats.Count = totalRows

	return errres
}

// deleteRemainingTargetData 删除目标表中剩余的数据
func (c *Controller) DeleteRemainingTargetData(ctx context.Context, rowsTarget common.RowsInterface, targetDB, targetSchema, targetTableName string, columns []common.KwdbColumn, IsSeries bool) (int64, error) {
	var deleted int64 = 0
	for {
		if common.IsStopped(ctx) {
			return deleted, nil
		}
		targetValue := make([]interface{}, len(columns))
		for i := 0; i < len(columns); i++ {
			targetValue[i] = &sql.NullString{}
		}
		err := rowsTarget.Scan(targetValue...)
		if err != nil {
			return deleted, err
		}
		_, err = c.writer.DeleteRecord(targetDB, targetSchema, targetTableName, columns, targetValue, "sync", IsSeries)
		if err != nil {
			return deleted, err
		}
		deleted++

		if !rowsTarget.Next() {
			break
		}
	}
	return deleted, nil
}

// insertRemainingSourceData 插入源表中剩余的数据
func (c *Controller) InsertRemainingSourceData(ctx context.Context, rowsSource common.RowsInterface, targetDB, targetSchema, targetTableName string, columns []common.KwdbColumn, IsSeries bool) (int64, error) {
	var added int64 = 0
	for {
		if common.IsStopped(ctx) {
			return added, nil
		}
		sourceValue := make([]interface{}, len(columns))
		for i := 0; i < len(columns); i++ {
			sourceValue[i] = &sql.NullString{}
		}
		err := rowsSource.Scan(sourceValue...)
		if err != nil {
			return added, err
		}
		_, err = c.writer.InsertRecords(targetDB, targetSchema, targetTableName, columns, [][]interface{}{sourceValue}, "sync", IsSeries)
		if err != nil {
			return added, err
		}
		added++

		if !rowsSource.Next() {
			break
		}
	}
	return added, nil
}
func (c *Controller) VerifyOneDB(ctx context.Context, stats *common.TaskStattistics, sourceDB, targetDB common.DatabaseConfig, IsSeries bool) error {
	tablesToSync, err := c.GetSyncTables(sourceDB)
	if err != nil {
		log.Println(err)
		return err
	}
	if targetDB.Includes != nil && len(targetDB.Includes) != len(tablesToSync) {
		log.Println("Source table count not match target table count")
	}
	for idx, sourceTable := range tablesToSync {
		if common.IsStopped(ctx) {
			break
		}
		var targetTable common.TableConfig
		if targetDB.Includes != nil {
			targetTable = targetDB.Includes[idx]
		} else {
			targetTable = common.TableConfig{Name: sourceTable.Name}
		}
		err := c.VerifyOneTable(ctx, stats, sourceDB.Name, sourceDB.Schema, sourceTable, targetDB.Name, targetDB.Schema, targetTable, IsSeries)
		if err != nil {
			log.Println(err)
		}
	}
	return nil
}
func (c *Controller) GetSyncTables(sourceDB common.DatabaseConfig) ([]common.TableConfig, error) {
	tables, err := c.reader.GetDatabaseTables(sourceDB.Name, sourceDB.Schema)
	if err != nil {
		log.Println(err)
		return nil, err
	}
	var tablesToSync []common.TableConfig
	//excludes
	excludeMap := make(map[string]bool)
	for _, exclude := range sourceDB.Excludes {
		excludeMap[exclude.Name] = true
	}

	// 过滤掉被排除的表
	filteredTables := []string{}
	for _, table := range tables {
		if !excludeMap[table] {
			filteredTables = append(filteredTables, table)
		}
	}
	tables = filteredTables
	//如果使用了includes，则只同步includes中的表
	if len(sourceDB.Includes) > 0 {
		tablesToSync = sourceDB.Includes
	} else {
		for _, table := range tables {
			tablesToSync = append(tablesToSync, common.TableConfig{Name: table})
		}
	}
	return tablesToSync, nil
}
func (c *Controller) SyncOneDB(ctx context.Context, stats *common.TaskStattistics, sourceDB, targetDB common.DatabaseConfig, IsSeries bool) error {
	if err := c.writer.CreateDatabase(targetDB.Name, IsSeries); err != nil {
		log.Println(err)
		return err
	}
	if err := c.writer.CreateSchema(targetDB.Name, targetDB.Schema, IsSeries); err != nil {
		log.Println(err)
		return err
	}
	tablesToSync, err := c.GetSyncTables(sourceDB)
	if err != nil {
		log.Println(err)
		return err
	}
	if targetDB.Includes != nil && len(targetDB.Includes) != len(tablesToSync) {
		log.Println("Source table count not match target table count")
		return errors.New("Source table count not match target table count")
	}
	// 创建信号量，限制最多parallelTable个goroutine并发执行
	semaphore := make(chan struct{}, c.parallelTable)
	var wg sync.WaitGroup
	var firstError error
	var mu sync.Mutex
	for idx, sourceTable := range tablesToSync {
		if common.IsStopped(ctx) {
			break
		}
		var targetTable common.TableConfig
		if targetDB.Includes != nil {
			targetTable = targetDB.Includes[idx]
		} else {
			targetTable = common.TableConfig{Name: sourceTable.Name}
		}
		// 获取信号量，支持 context 取消
		select {
		case semaphore <- struct{}{}:
			// 成功获取信号量
			wg.Add(1)

			go func(sourceTable common.TableConfig, targetTable common.TableConfig) {
				defer func() {
					<-semaphore
					wg.Done()
				}()

				err := c.SyncOneTable(ctx, stats, sourceDB.Name, sourceDB.Schema, sourceTable,
					targetDB.Name, targetDB.Schema, targetTable, IsSeries)
				if err != nil {
					log.Println(err)
					mu.Lock()
					if firstError == nil {
						firstError = err
					}
					mu.Unlock()
				}
			}(sourceTable, targetTable)

		case <-ctx.Done():
			// context 被取消，等待已启动的 goroutine 完成
			wg.Wait()
			return ctx.Err()
		}
	}
	// 等待所有goroutine完成
	wg.Wait()

	return firstError
}
func (c *Controller) CommonCheck(task *common.MigrateTask) error {
	var err error = nil
	if c.reader == nil {
		c.reader, err = readers.CreateReader(task.Source.DBCon)
	}
	if err != nil {
		log.Println(err)
		return err
	}
	if c.writer == nil {
		c.writer, err = writers.CreateWriter(task.Target.DBCon)
	}
	if err != nil {
		log.Println(err)
		return err
	}
	//验证数据库账户
	err = c.reader.GetDB().Ping()
	if err != nil {
		log.Println(err)
		return err
	}
	err = c.writer.GetDB().Ping()
	if err != nil {
		log.Println(err)
		return err
	}
	if len(task.Source.Databases) != len(task.Target.Databases) {
		log.Println("Source database count not match target database count")
		return err
	}
	return nil
}

// getTargetDatabaseInfo 根据源数据库信息获取目标数据库信息
func (c *Controller) GetTargetDatabaseInfo(sourceDBName, sourceSchemaName, sourceTableName string) (targetDBName, targetSchemaName, targetTableName string) {
	// 遍历任务中的源数据库和目标数据库配对
	for i, sourceDB := range c.task.Source.Databases {
		targetDB := c.task.Target.Databases[i]

		// 检查数据库名是否匹配
		if sourceDB.Name != sourceDBName {
			continue
		}

		// 如果没有配置includes，则目标和源相同
		if len(targetDB.Includes) == 0 {
			return targetDB.Name, targetDB.Schema, sourceTableName
		}

		// 如果配置了includes，需要查找对应关系
		for j, sourceTable := range sourceDB.Includes {
			if sourceTable.Name == sourceTableName {
				// 确保目标数据库也有对应的表配置
				if j < len(targetDB.Includes) {
					targetTable := targetDB.Includes[j]
					return targetDB.Name, targetDB.Schema, targetTable.Name
				}
			}
		}
	}

	// 未找到匹配项
	return sourceDBName, sourceSchemaName, sourceTableName
}
func (c *Controller) ReplicateRecords(dbName, schemaName, tableName string, columns []common.KwdbColumn, values [][]interface{}, replicateType string) (uint64, error) {

	//得到writer的对应的数据库，模式和表名
	targetDBName, targetSchemaName, targetTableName := c.GetTargetDatabaseInfo(dbName, schemaName, tableName)
	sourceType := "replicate"
	if c.reader.GetDataBaseType() == "postgresql" {
		sourceType = "sync"
	}
	return c.writer.ReplicateRecords(targetDBName, targetSchemaName, targetTableName, columns, values, sourceType, replicateType)
}
func (c *Controller) DoMigrateTask(ctx context.Context, task *common.MigrateTask, stats *common.TaskStattistics, batchSize, parallelTable, parallelInsert int) error {

	c.batchSize = batchSize
	if c.batchSize < 100 {
		c.batchSize = 100
	}
	c.parallelTable = parallelTable
	if c.parallelTable < 1 {
		c.parallelTable = 1
	}
	c.parallelInsert = parallelInsert
	if c.parallelInsert < 1 {
		c.parallelInsert = 1
	}

	c.task = task
	stats.StartTime = time.Now()
	stats.TaskName = task.Name
	stats.TaskType = "migrate"
	err := c.CommonCheck(task)
	if err != nil {
		return err
	}
	var startPos common.ReplicateStartPos
	if strings.Contains(task.Mode, "inc") {
		startPos, err = c.reader.GetStartPos()
		if err != nil {
			log.Println(err)
			return err
		}
	}

	c.reader.SetDataBases(task.Source.Databases)
	if strings.Contains(task.Mode, "inc") {
		c.reader.SetReplicateHandler(c)
	}
	for i, sourceDB := range task.Source.Databases {
		targetDB := task.Target.Databases[i]
		err = c.SyncOneDB(ctx, stats, sourceDB, targetDB, sourceDB.IsSeries)
		if err != nil {
			log.Println(err)
		}
	}
	stats.EndTime = time.Now()
	if err == nil && strings.Contains(task.Mode, "inc") {
		//显示全量迁移信息
		common.PrintStatistics(stats)
		color.New(color.FgGreen, color.Bold).Println("开始增量迁移Start replication")
		c.reader.StartReplicate(ctx, startPos)
	}
	if c.reader != nil {
		c.reader.Close()
	}
	if c.writer != nil {
		c.writer.Close()
	}

	//show the statistics
	//common.PrintStatistics(stats)
	return nil
}

// 验证数据库数据是否一致
func (c *Controller) DoVerifyTask(ctx context.Context, task *common.MigrateTask, stats *common.TaskStattistics) error {
	//var stats common.TaskStattistics
	stats.StartTime = time.Now()
	stats.TaskName = task.Name
	stats.TaskType = "verify"

	log.Println("Verify task")
	err := c.CommonCheck(task)
	if err != nil {
		return err
	}
	for i, sourceDB := range task.Source.Databases {
		targetDB := task.Target.Databases[i]
		err = c.VerifyOneDB(ctx, stats, sourceDB, targetDB, sourceDB.IsSeries)
		if err != nil {
			log.Println(err)
		}
	}
	stats.EndTime = time.Now()
	if c.reader != nil {
		c.reader.Close()
	}
	if c.writer != nil {
		c.writer.Close()
	}
	//show the statistics
	//common.PrintStatistics(stats)
	return nil
}
